package com.qqt.csr.im.socketio;

import cn.hutool.core.date.DateUtil;
import com.corundumstudio.socketio.AckRequest;
import com.corundumstudio.socketio.SocketIOClient;
import com.corundumstudio.socketio.annotation.OnConnect;
import com.corundumstudio.socketio.annotation.OnDisconnect;
import com.corundumstudio.socketio.annotation.OnEvent;
import com.qqt.csr.common.utils.*;
import com.qqt.csr.im.dto.MsgAccountDTO;
import com.qqt.csr.im.entity.Connection;
import com.qqt.csr.im.entity.Conversation;
import com.qqt.csr.im.entity.RoomMember;
import com.qqt.csr.im.entity.mq.MqMessage;
import com.qqt.csr.im.entity.msg.Message;
import com.qqt.csr.im.entity.msg.TextContent;
import com.qqt.csr.im.enums.MsgFlowTypeEnum;
import com.qqt.csr.im.enums.MsgTypeEnum;
import com.qqt.csr.im.mq.SendMessageListener;
import com.qqt.csr.im.service.AccountService;
import com.qqt.csr.im.service.ConnectionService;
import com.qqt.csr.im.service.EventMsgProducerService;
import com.qqt.csr.im.service.RoomMemberService;
import com.qqt.csr.im.service.mongo.ConversationService;
import com.qqt.csr.im.service.mongo.MessageService;
import com.qqt.csr.im.service.sensitiveword.SensitiveWordService;
import com.qqt.csr.im.socketio.constants.EventsConstants;
import com.qqt.csr.im.socketio.utils.SocketIOLocalCacheUtil;
import com.qqt.csr.im.socketio.utils.SocketIOUtil;
import com.qqt.csr.im.utils.DeviceCacheKeyUtil;
import com.qqt.csr.im.utils.ServiceHelper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @Author: yang hao
 * @Date: 2024/10/15 15:07
 * @Description:
 **/
@Slf4j
@Component
@DependsOn("socketIOLocalCacheUtil")
public class SocketIOEventHandler {
    @Autowired
    private RedissonClient redissonClient;
    @Autowired
    private EventMsgProducerService producerService;
    @Autowired
    private ConnectionService connectionService;
    @Autowired
    private AccountService accountService;
    @Autowired
    private MessageService messageService;
    @Autowired
    private ConversationService conversationService;
    @Autowired
    private SendMessageListener sendMessageListener;
    @Autowired
    private TransactionUtil transactionUtil;
    @Autowired
    private RedissonLockUtil redissonLockUtil;
    @Autowired
    private RoomMemberService roomMemberService;
    @Autowired
    private SensitiveWordService sensitiveWordService;

    /**
     * 连接事件监听
     *
     * @param client {@link SocketIOClient} websocket 客户端
     */
    @OnConnect
    public void onConnect(SocketIOClient client) {
        String deviceKey = DeviceCacheKeyUtil.getKey(client);
        SocketIOClient existClient = SocketIOLocalCacheUtil.getClient(deviceKey);
        if (existClient != null && existClient.equals(client) && existClient.isChannelOpen()) {
            // socketIO框架的一个bug，客户端连接成功后，会回调两次OnConnect事件，这里做个判断过滤最后一次回调
            log.info("客户端:{}已连接，但连接已存在，直接返回！key：{}", client.getSessionId(), deviceKey);
            return;
        }

        String lockKey = String.format(CacheKeyUtil.IM.UPDATE_CONNECTION_LOCK, deviceKey);
        redissonLockUtil.distributedLock(() -> {

            clearOldSocketIOInOtherNode(client, deviceKey);

            if (!client.isChannelOpen()) {
                log.info("客户端:{}已被断开连接，直接返回！key：{}", client.getSessionId(), deviceKey);
                return;
            }
            // client存入本地缓存
            Connection connection = connectionService.connect(client);
            SocketIOLocalCacheUtil.setClient(deviceKey, client);
            client.set("connection", connection);
            log.info("客户端:{}已连接,key={},serviceId:{}", client.getSessionId(), deviceKey, ServiceHelper.INSTANCE.getServiceNode().getId());
        }, lockKey, 60, 30, TimeUnit.SECONDS);
    }

    /**
     * 通知其他节点，当前连接已经在本节点建立长连接，可以清理之前已缓存的无效连接信息
     * 主要应对非正常的断开连接
     *
     * @param client
     * @param deviceKey
     */
    private void clearOldSocketIOInOtherNode(SocketIOClient client, String deviceKey) {
        // 如果本机有老的连接，那就先关闭老的，以便后续使用新的
        SocketIOUtil.tryCloseOldSocket(client, deviceKey);
        // 通知其他节点，当前连接已经在本节点建立长连接，可以清理之前已缓存的无效连接信息
        Long currentServiceNodeId = ServiceHelper.INSTANCE.getServiceNode().getId();
        MqMessage data = MqMessage.of(currentServiceNodeId, deviceKey, null, null);
        producerService.sendEventMsg(EventsConstants.CLEAR_LOCAL_CACHE_DISCONNECT, data);
    }

    /**
     * 正常断开连接事件监听
     *
     * @param client {@link SocketIOClient} websocket 客户端
     */
    @OnDisconnect
    public void onDisconnect(SocketIOClient client) {
        String deviceKey = DeviceCacheKeyUtil.getKey(client);
        String lockKey = String.format(CacheKeyUtil.IM.UPDATE_CONNECTION_LOCK, deviceKey);
        redissonLockUtil.distributedLock(() -> {
            Connection connection = client.get("connection");
            if (connection == null) {
                log.info("非法链接断开回调，key:{},sessionId:{}", deviceKey, client.getSessionId());
                return;
            }

            SocketIOClient cacheClient = SocketIOLocalCacheUtil.getClient(deviceKey);
            if (!client.equals(cacheClient)) {
                log.info("客户端:{}又重新创建新连接，直接返回！key：{}", client.getSessionId(), deviceKey);
                return;
            }
            // 处理断开流程
            connectionService.disconnect(cacheClient);
            SocketIOLocalCacheUtil.remove(deviceKey);
            log.info("客户端:{}断开连接,key={},serviceId:{}", client.getSessionId(), deviceKey, connection.getServiceId());
        }, lockKey, 60, 30, TimeUnit.SECONDS);
    }

    /**
     * 自定义事件监听
     *
     * @param client  {@link SocketIOClient} websocket 客户端
     * @param request {@link AckRequest} ack请求
     * @param message {@link Message} 请求数据
     */
    @OnEvent(value = EventsConstants.C2S_MESSAGE_EVENT)
    public void onEvent(SocketIOClient client, AckRequest request, Message message) {
        Conversation conversation = conversationService.getById(message.getConversationId());
        if (Objects.isNull(conversation) || !conversation.isNormal()) {
            log.warn("会话已关闭，message:{}", JsonUtil.toJson(message));
            callbackIllegalStateMsg(client, message.getConversationId(), "会话已结束");
            return;
        }

        List<RoomMember> roomMemberList = roomMemberService.queryMemberListBy(Long.parseLong(conversation.getRoomId()));
        if (CollectionUtils.isEmpty(roomMemberList)) {
            log.warn("房间不存在，conversationId:{},message:{}", conversation.getRoomId(), JsonUtil.toJson(message));
            callbackIllegalStateMsg(client, message.getConversationId(), "Room is not exist!");
            return;
        }

        Long senderAccountId = client.get("accountId");
        if (roomMemberList.stream().noneMatch(roomMember -> senderAccountId.equals(roomMember.getAccountId()))) {
            log.warn("非法消息，消息推送人不在当前会话的房间中，message:{}", JsonUtil.toJson(message));
            callbackIllegalStateMsg(client, message.getConversationId(), "Illegal push!");
            return;
        }

        String tenantId = client.get("tenantId");
        // 填充必要的默认数据
        populateDefaultMsgData(message, senderAccountId, conversation, tenantId);
        //保存副本消息
        List<Message> pushMsgList = messageService.saveDuplicateMsg(senderAccountId, message, roomMemberList);

        // 寻找房间中所有成员的所有设备连接，用于同步消息
        List<Long> accountIdList = roomMemberList.stream().map(RoomMember::getAccountId).distinct().collect(Collectors.toList());
        Map<Long, List<Connection>> connectionMap = connectionService.getOnlineByAccountId(accountIdList);
        // 推送消息到接收方的设备连接上
        for (Message pshMsg : pushMsgList) {
            Long recipientAccountId = Long.parseLong(pshMsg.getRecipient().getId());
            List<Connection> recipientConnections = connectionMap.get(recipientAccountId);
            if (CollectionUtils.isEmpty(recipientConnections)) {
                log.warn("im账号无连接信息，message:{}，accountId:{}", JsonUtil.toJson(pshMsg), recipientAccountId);
                continue;
            }

            if (pshMsg.getContent() instanceof TextContent textContent && MsgTypeEnum.TextElem.equals(pshMsg.getType())) {
                // 敏感词替换，推送到接收方展示
                String replace = sensitiveWordService.replace(textContent.getText(), pshMsg.getSensitiveWords());
                textContent.setText(replace);
            }

            for (Connection connection : recipientConnections) {
                // 异步推送消息内容到连接设备上
                MqMessage mqData = MqMessage.of(connection.getServiceId(), DeviceCacheKeyUtil.getKey(connection), pshMsg, null);
                producerService.sendEventMsg(EventsConstants.S2S_MESSAGE_EVENT, mqData);
            }
        }

        //原消息内容回调第三方
        callBackMsgToThird(message, senderAccountId);

    }

    /**
     * 返回非法推送消息
     *
     * @param client
     * @param conversationId
     * @param tip
     */
    private void callbackIllegalStateMsg(SocketIOClient client, String conversationId, String tip) {
        Long senderAccountId = client.get("accountId");
        Message closedConversationMsg = Message.builder()
                .conversationId(conversationId)
                .recipient(MsgAccountDTO.builder().id(senderAccountId.toString()).build())
                .type(MsgTypeEnum.SystemElem)
                .content(TextContent.of(tip))
                .flow(MsgFlowTypeEnum.IN)
                .build();
        client.sendEvent(EventsConstants.S2C_MESSAGE_EVENT, closedConversationMsg);
    }

    /**
     * 填充默认消息数据
     *
     * @param message
     * @param senderAccountId
     * @param conversation
     */
    private void populateDefaultMsgData(Message message, Long senderAccountId, Conversation conversation, String tenantId) {
        message.setTenantId(tenantId);
        message.setSeq(SnowflakeUtil.nextIdStr());
        message.setSender(MsgAccountDTO.builder().id(senderAccountId.toString()).build());
        message.setRoomId(conversation.getRoomId());
        message.setCreateTime(DateUtil.now());
        message.setUpdateTime(message.getCreateTime());
        message.convertContent();
    }

    /**
     * 原消息内容回调第三方
     * @param message
     * @param senderAccountId
     */
    private void callBackMsgToThird(Message message, Long senderAccountId) {
        Message copy= new Message();
        copy.setSeq(message.getSeq());
        copy.setConversationId(message.getConversationId());
        copy.setContent(message.getContent());
        copy.setType(message.getType());
        copy.setArtificialExt(message.getArtificialExt());
        copy.setCreateTime(message.getCreateTime());
        copy.setUpdateTime(message.getUpdateTime());
        copy.setRoomId(message.getRoomId());
        copy.setTenantId(message.getTenantId());
        copy.setSender(MsgAccountDTO.of(accountService.getById(senderAccountId)));
        producerService.sendCallbackMsg(copy);
    }


}
